package com.smrtcold.dev.config;


import com.smrtcold.dev.init.SystemInit;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * @author WJH
 * @date 2019/4/1110:19
 */
@Log4j2
@Component
@Configuration
public class MqttBase {


    private final String CID = UUID.randomUUID().toString().substring(32);
    //
    public ExecutorService DATAPOOL = Executors.newCachedThreadPool();

    @Value("${sys.apid}")
    private String apid;
    @Value("${spring.mqtt.url}")
    private String[] url;
    @Value("${spring.mqtt.username}")
    private String username;
    @Value("${spring.mqtt.password}")
    private String password;
    @Value("${spring.mqtt.timeout}")
    private int timeout;
    @Value("${spring.mqtt.keepalive}")
    private int keepalive;
    @Value("${spring.mqtt.clientid}")
    private String clientid;
    @Value("${sys.subData}")
    private boolean subData;
    @Value("${spring.mqtt.prefix}")
    private String prefix;


    @Autowired
    private SystemInit sender;

    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions option = new MqttConnectOptions();
        //设置是否清空session,false表示服务器会保留客户端的连接记录，true表示每次连接到服务器都以新的身份连接
        option.setCleanSession(true);
        //自动断开重连
        option.setAutomaticReconnect(true);
        option.setServerURIs(url);
        //设置连接的用户名
        option.setUserName(username);
        //设置连接的密码
        option.setPassword(password.toCharArray());
        //设置超时时间 单位为秒
        option.setConnectionTimeout(timeout);
        //设置会话心跳时间 单位为秒 服务器会每隔(1.5*keepTime)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        option.setKeepAliveInterval(keepalive);
        option.setConnectionTimeout(10);     // 连接超时时间(秒)
        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
        // option.setWill(TOPICUtil.WILL, "close".getBytes(), 1, true);
        return option;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }


    @EventListener
    public void handleConnectionFailed(MqttConnectionFailedEvent event) {
        log.error("MQTT连接失败: " + event.getCause().getMessage());
        // 这里可以添加你的连接失败处理逻辑
//                todo 延时---暂停所有任务----并交出控制权---等待连接成功---重新启动任务
        SystemInit.addCache("MQTT_STATUS");
    }


    // -----------------------------------------------------------------设备发送数据--------------------------------------------------------------------
    @Bean
    @ServiceActivator(inputChannel = "DevOperChannel")
    public MessageHandler mqttOutbound(MqttPahoClientFactory mqttPahoClientFactory) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientid + this.apid + "_MD_"+CID, mqttPahoClientFactory);
        messageHandler.setAsync(false);
        return messageHandler;
    }

    @Bean
    public MessageChannel DevOperChannel() {
        return new DirectChannel();
    }
//    //MQTT 协议===========================================================================================================================================
    @Bean
    public MessageChannel xyChannel() {
        return new DirectChannel();
    }
    //------------------------------------------------------------------------------------------------------------------------------------

    @Bean(name = "xyRes")
    public MessageProducer xyData(MqttPahoClientFactory mqttPahoClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientid + this.apid + "_HM_OP"+CID, mqttPahoClientFactory, this.prefix +"/" + apid + "/#");
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        adapter.setOutputChannel(xyChannel());
//        adapter.addTopic(  "HHH/10000/MonitorData",1);
        return adapter;
    }


    @Bean
    @ServiceActivator(inputChannel = "xyChannel")
    public MessageHandler xyHandler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                DATAPOOL.submit(() -> {
                    try {
                        String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                        String[] topics = topic.split("/");//HPI/apid/devid/oper
                        String resData = message.getPayload().toString();
                        String apid = topics[1], devid = topics[2], top = topics[3];
                        switch (top) {
                            //操作
                            //收到设备配置
                            case "conf":
//                                HPI/apid/all/conf/**      HPI/apid/devid/[add,update,delete,stop]
                                sender.initDevConf(apid, devid, "all".equals(top) ? "all" : topics[topics.length - 1], resData);
                                break;
                            //执行任务调度
                            case "task":
//                                HPI/apid/all/task  HPI/apid/taskid/task/[add,update,delete,run]  taskid：当为定时任务时为ID/其他任务为名称
                                sender.initTask(apid, devid, "all".equals(devid) ? top : topics[topics.length - 1], resData);
                                break;
                            case "data":
                                //HPI/apid/devid/data
                                sender.getData(apid, devid, resData,"data");
                                break;
                            case "mtdata":
                                sender.getData(apid, devid, resData,"mtdata");
                                 break;
                            // //HPI/apid/all/status
                            case "status":
                                sender.getDevStatus(apid,devid,"devStatus");
                                break;
//                            case "status":
//                               // HPI/apid/status/devStatus
//
//                                break;
                            case "read":
                                //读取数据 {环境}/apid/{devid}/read/{uuid}
                                sender.read(apid, devid, resData,topics[4]);
                                break;
                            case "write":
                                //操作结果 {环境}/apid/{devid}/write/{uuid}
                                sender.write(apid, devid, resData,topics[4]);

                                break;
                            case "sys":
//                                HPI/%s/{status||conf}/sys  {0/1}
                                if ("status".equals(devid)) {
                                    sender.setISRUN("1".equals(resData), 1);
                                } else if ("conf".equals(devid)) {
                                    sender.initSysConf(resData);
                                }
                                break;


                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        };
    }


    @MessagingGateway(defaultRequestChannel = "DevOperChannel")
    public interface DevOperation {

        void push(@Header(MqttHeaders.TOPIC) String topic, String data);

        void push(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String data);
    }


}

